package mgoasync

import (
	"container/list"
	"context"
	"fmt"
	"time"

	"gitee.com/terender/sfw/rlog"
	"go.mongodb.org/mongo-driver/mongo"
)

// 批量更新后台任务
// 用于数据总个数不经常变化但每个数据的内容频繁发生改变的数据批量更新MongoDB
// 可以是全量更新也可以是增量更新
// 由于数据频繁发生改变，一般在打包同时将数据序列化，减少数据拷贝的次数及锁的竞争
type UpdateTask struct {
	*Task
}

// 数据实体的抽象结构
type Entity interface {
	// 数据对应的文档的筛选器，一般是数据对应文档在数据库中的唯一标识
	Selector() interface{}
	// 数据实体的打包方法，用于生成回写数据库的更新数据
	Pack() interface{}
}

// 数据实体加入更新后台任务后的元素对象，包含其在缓存链表中的节点元素
type Element struct {
	entity *Entity
	le     *list.Element
}

// 返回任务中元素内包含的数据实体
func (e *Element) Data() *Entity { return e.entity }

// 构造一个数据库更新的后台任务，默认间隔时间 1 分钟
func NewUpdateTask(coll *mongo.Collection) *UpdateTask {
	return &UpdateTask{Task: newTask(coll)}
}

// 任务开始执行
func (task *UpdateTask) Begin() error {
	if err := task.begin(); err != nil {
		return err
	}
	go task.writeLoop()
	return nil
}

// 向后台任务的更新数据实体池中添加一个数据实体对象
func (task *UpdateTask) AddEntity(entity Entity) *Element {
	ele := &Element{
		entity: &entity,
	}
	task.Lock()
	defer task.Unlock()
	ele.le = task.queue.PushBack(entity)
	return ele
}

// 从后台任务的更新数据实体池中移除一个数据实体对象
func (task *UpdateTask) RemoveEntity(ele *Element) {
	task.Lock()
	defer task.Unlock()
	task.queue.Remove(ele.le)
}

// 复制数据实体链表，确保读锁时间最短
func (task *UpdateTask) cloneQueue() *list.List {
	task.RLock()
	defer task.RUnlock()
	q := list.New()
	q.PushBackList(task.queue)
	return q
}

// 更新数据库线程循环
func (task *UpdateTask) writeLoop() {
	defer func() {
		task.setStatus(StatusIdle)
		close(task.closed)
	}()
	name := fmt.Sprintf("%v.%v", task.mgoColl.Database().Name(), task.mgoColl.Name())

	for {
		// 不是退出状态，也没有失败后等待重写的数据，此时需要等待时间间隔
		if task.pendingDatas == nil && !task.isQuiting() {
			select {
			case <-task.ticker.C:
			case <-task.closing:
			}
		}

		// 写入统计数据
		act := &Act{OP: OPUpdate, StartTime: time.Now()}

		// 如果待更新数据为空，则遍历全部数据实体，对其进行打包操作，生成待更新的数据列表
		// 如果待更新数据不为空，说明上次更新失败了，此时不再进行新的打包操作，而是将上次的数据重新写入
		var datas []mongo.WriteModel
		if task.pendingDatas == nil {
			queue := task.cloneQueue()
			for e := queue.Front(); e != nil; e = e.Next() {
				data := e.Value.(Entity)
				patch := data.Pack()
				if patch == nil {
					continue
				}
				datas = append(datas, mongo.NewUpdateOneModel().SetFilter(data.Selector()).SetUpdate(patch).SetUpsert(true))
			}
		} else {
			datas = task.pendingDatas.([]mongo.WriteModel)
			task.pendingDatas = nil
		}
		// 打包完成，记录数据长度和打包时间
		act.PackTime = time.Now()
		act.Count = len(datas)

		// 没有产生任何待更新数据，此次更新操作直接完成，准备开始下一次
		if len(datas) == 0 {
			act.FinishTime = time.Now()
			// 如果已经处于退出过程中, 则表示全部数据回写完成，可以退出
			if task.isQuiting() {
				rlog.Infof(`异步批量回写更新 %v.%v 完成，更新了 %v 个文档`, task.mgoColl.Database().Name(), task.mgoColl.Name(), len(datas))
				break
			}
			continue
		}
		addStatistics(name, act)

		// 将数据批量更新至 MongoDB
		_, err := task.mgoColl.BulkWrite(context.TODO(), datas)
		act.FinishTime = time.Now()

		// 如果写入出错，则等待3秒后立即尝试重写
		if err != nil {
			act.Err = err
			rlog.Errorf(`异步批量回写更新 %v.%v 遇到错误 %v`, task.mgoColl.Database().Name(), task.mgoColl.Name(), err)
			task.pendingDatas = datas
			time.Sleep(time.Second * 3)
		}
	}
}
